#include "Reactor.h"

#include <utility>
#include "ReactorTask.h"

#if defined(WIN32) || defined(_WIN32)
#include "../interface/Select.h"
#else
#include "../interface/Epoll.h"
#endif

using namespace std;
using namespace network;
using namespace threadpool;

Reactor::Reactor(shared_ptr<threadpool::ThreadPool> pool) : pool_(std::move(pool)), stop_(false) {
#if defined(WIN32) || defined(_WIN32)
    io_multiplexer_ = make_shared<Select>();
#else
    io_multiplexer_ = make_shared<Epoll>();
#endif
    pool_->add(1);
}

bool Reactor::registerHandler(int fd, EventType event_type, const Event::Handler &handler, void *context) {
    event_map_.emplace(fd, make_shared<Event>(fd, handler, context));
    return io_multiplexer_->add(fd, event_type);
}

bool Reactor::changeEventType(int fd, EventType event_type) {
    return io_multiplexer_->mod(fd, event_type);
}

void Reactor::removeHandler(int fd) {
    io_multiplexer_->del(fd);
    ready_event_set_mutex_.lock();
    std::shared_ptr<Event> ev = event_map_.at(fd);
    if (ready_event_set_.count(ev.get()) == 1) {
        ready_event_set_.erase(ev.get());
    }
    event_map_.erase(fd);
    ready_event_set_mutex_.unlock();
}

void Reactor::addEvent(int fd, EventType event_type) {
    Event *ev = event_map_.at(fd).get();
    ev->addEventLock(event_type);
    ready_event_set_mutex_.lock();
    ready_event_set_.emplace(ev);
    ready_event_set_mutex_.unlock();
    if (!ev->isActive()) {
        ev->setReady();
    }
}

void Reactor::loop() {
    stop_.store(false);
    size_t remaining_threads = 0;
    while (!stop_.load()) {
        try {
            io_multiplexer_->dispatch(this);
        }
        catch (const exception &e) {
            stop_.store(true);
            break;
        }
        remaining_threads = pool_->threadCount() - pool_->workThreadCount();
        ready_event_set_mutex_.lock();
        auto it = ready_event_set_.begin();
        auto end = ready_event_set_.end();
        while (remaining_threads > 0 && it != end) {
            Event *ev = static_cast<Event*>(*it);
            if (ev->isReady()) {
                EventType type = ev->popLock();
                ev->setActive();
                pool_->submit(make_shared<ReactorTask>(
                    std::bind(ev->getHandler(), ev->getFd(), type, ev->getContext()),
                    ev,
                    this));
                remaining_threads--;
                ++it;
            }
            else {
                ++it;
            }
        }
        ready_event_set_mutex_.unlock();
    }
}
